package com.atguigu.chapter11;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink01_Table_BaseUse {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        );

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. 用环境，将流转换成表
        Table table = tEnv.fromDataStream(stream);
        // 输出表的详细信息
        table.printSchema();

        // 2. 对表对象进行查询
        // select * from t where id='sensor_1'
        Table result = table
//                .where(Expressions.$("id").isEqual("sensor_1"))  // 每次写Expressions太麻烦了
//                .select(Expressions.$("id"),Expressions.$("vc"));
                .where($("id").isEqual("sensor_1"))  // 静态导入Expressions.$
                .select($("id"),$("vc"));

        // 3. 再把动态表转换成流，注意泛型填的是流里面放的数据类型
//        DataStream<WaterSensor> rowDataStream = tEnv.toDataStream(result,WaterSensor.class);
//        DataStream<Row> rowDataStream = tEnv.toDataStream(result,Row.class);  // 这样就不会出现列类型不一样得了，row是通用的
        DataStream<Row> rowDataStream = tEnv.toAppendStream(result,Row.class);  //只有新增还有这种写法


        // 4. 输出结果
        rowDataStream.print();


        // ***** 注意一定要提交，要不然没有输出结果
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
